package com.baitengsoft.patentservicemonitor.job.cloud.download;

import com.alibaba.fastjson.JSON;
import com.aliyun.oss.OSSClient;
import com.aliyun.oss.common.utils.IOUtils;
import com.aliyun.oss.model.ListObjectsRequest;
import com.aliyun.oss.model.OSSObject;
import com.aliyun.oss.model.OSSObjectSummary;
import com.aliyun.oss.model.ObjectListing;
import com.baitengsoft.patentservicemonitor.api.service.WebPlatformService;
import com.baitengsoft.patentservicemonitor.job.model.PacPatentinfoModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 数据初始化
 *
 * @author wangze
 * @date 2018-02-13 14:37.
 **/
@Component
public class InitAliYunDataService extends AbstractInitDataService {


    @Autowired
    WebPlatformService webPlatformService;

    @Override
    public String getFolder(int platformId) {

        return "data/" + platformId + "/";
    }

    @Override
    protected PacPatentinfoModel dealData(Integer pid, String json, String key) {
        PacPatentinfoModel p = super.dealData(pid, json, key);
        pacPatentinfoService.insertByBatch(Arrays.asList(p));
        return p;
    }


    /**
     * 多线程获取oss 数据， 事务批量插入数据，不用于更新
     *
     * @param pid
     */
    public void batchInsert(Integer pid) {
        ConcurrentLinkedQueue<PacPatentinfoModel> queue = new ConcurrentLinkedQueue<>();
        ExecutorService executorService = Executors.newFixedThreadPool(this.maxThreads);
        OSSClient ossClient = this.getOssClient();
        String nextMarker = null;

        ListObjectsRequest listObjectsRequest = new ListObjectsRequest(config.getAliyunBucketName());
        listObjectsRequest.setPrefix(getFolder(pid));
        ObjectListing objectListing;

        long startTime = System.currentTimeMillis();
        do {
            objectListing = ossClient.listObjects(listObjectsRequest.withMarker(nextMarker).withMaxKeys(this.getMaxFetchDataSize()));
            List<OSSObjectSummary> sums = objectListing.getObjectSummaries();
            if (sums.size() == 0) {
                continue;
            }
            for (OSSObjectSummary s : sums) {
                executorService.submit(() -> {
                    try {
                    OSSObject ossObject = ossClient.getObject(config.getAliyunBucketName(), s.getKey());
                        String json = IOUtils.readStreamAsString(ossObject.getObjectContent(), "utf-8");
                        PacPatentinfoModel p = JSON.parseObject(json.replace("\"class\"" , "\"clazz\""), PacPatentinfoModel.class);
                        if (p == null) {
                            System.out.println(getName(ossObject.getKey()));
                            return;
                        }
                        p.setPlatformid(pid);
                        queue.add(p);
                        System.out.println(p.getAn() + "    " + (System.currentTimeMillis() - startTime) / 1000);
                    } catch (Exception e) {
                        System.out.println(e);
                        System.out.println("失败 " + s.getKey());
                        e.printStackTrace();
                        throw new RuntimeException(e);
                    }
                });
            }
            nextMarker = objectListing.getNextMarker();
        } while (objectListing.isTruncated());

        executorService.shutdown();

        Connection conn = null;
        PreparedStatement pstm = null;

        try {
            long startTime2 = System.currentTimeMillis();
            conn = dataSource.getConnection();
            conn.setAutoCommit(false);
            String sql = " insert into pac_patentinfo " +
                    "   (" +
                    "   `PlatformID`,`PatNum`,`PatName`,`PatTypeName`,`LawState`,\n" +
                    "   `PatAppDate`,`PatApproveDate`,`PublicationDate`,`ExpiryDate`,`AgencyName`,\n" +
                    "   `InventPerson`,`ApplyPerson`,`ApplyPersonLast`,`PatAddress`,`PatAddressPrevious`,\n" +
                    "   `PatAddressLast`,`PatSubExamDate`,`IPCId`,`EffectReason`,`ApplyPersonType`,\n" +
                    "   `DateLast`,`ChinaIPCID`,`ApplyPersonPrevious`,`ApplyPersonTypePrevious`,`IsRollIn`, `CDN`,`PDN`\n" +
                    "   )" +
                    "values(?,?,?,?,?" +
                    ",?,?,?,?,?" +
                    ",?,?,?,?,?" +
                    ",?,?,?,?,?" +
                    ",?,?,?,?,?,?,?)";
            pstm = conn.prepareStatement(sql);
            while (!queue.isEmpty() || !executorService.isTerminated()) {
                PacPatentinfoModel p = queue.poll();
                if (p == null) {
                    continue;
                }

                pstm.setLong(1, p.getPlatformid());
                pstm.setString(2, p.getAn());
                pstm.setString(3, p.getTi());
                pstm.setInt(4, p.getClazz());
                pstm.setInt(5, p.getLawstate());
                pstm.setString(6, p.getAd());
                pstm.setString(7, p.getApd());
                pstm.setString(8, p.getPd());
                pstm.setString(9, p.getIvd());
                pstm.setString(10, p.getCagc());
                pstm.setString(11, p.getIn1());
                pstm.setString(12, p.getPa1());
                pstm.setString(13, p.getCpa1());
                pstm.setString(14, p.getAa());
                pstm.setString(15, p.getPaa());
                pstm.setString(16, p.getCaa());
                pstm.setString(17, p.getSed());
                pstm.setString(18, p.getIc1());
                pstm.setString(19, p.getLsn2());
                pstm.setInt(20, p.getApplyPersonType());
                pstm.setString(21, p.getTfd());
                // pstm.setString(22, p.getChinaIPCID());
                pstm.setString(23, p.getPpa1());
                pstm.setInt(24, p.getApplyPersonTypePrevious());
                pstm.setBoolean(25, p.getIsRollIn());
                pstm.setString(26,p.getCdn());
                pstm.setString(27,p.getPdn());
                pstm.addBatch();
            }
            pstm.executeBatch();
            conn.commit();
            long entTime2 = System.currentTimeMillis();
            System.out.println("插入数据库花费时间：" + (entTime2 - startTime2) / 1000 + "秒");
            webPlatformService.updatePatentNum(pid);

        } catch (Exception ex) {
            System.out.println(ex);
        } finally {

            if (pstm != null) {
                try {
                    pstm.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                    throw new RuntimeException(e);
                }
            }
            if (conn != null) try {
                conn.close();
            } catch (SQLException e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
            ossClient.shutdown();

        }

    }


    @Override
    public void fetchDataSuccessCallback(Integer pid) {

    }
}
